package com.suning.service.impl;

import com.suning.constant.Constants;
import com.suning.entity.EsBasePojo;
import com.suning.es.EsConfig;
import com.suning.service.BaseService;
import com.suning.service.ElasticSearchService;
import com.suning.utils.EsUtils;
import com.suning.utils.JestUtils;
import com.suning.utils.QueryStringJSONBuilder;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.Search;
import io.searchbox.core.SearchResult;
import io.searchbox.core.SearchScroll;
import io.searchbox.params.Parameters;
import net.sf.json.JSONObject;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import static com.suning.constant.Constants.MAX_LIST_CAPACITY;
import static com.suning.constant.Constants.SCROLL_TIMEOUT;
import static io.searchbox.params.SearchType.DFS_QUERY_THEN_FETCH;
import static io.searchbox.params.SearchType.QUERY_THEN_FETCH;

/**
 * 〈一句话功能简述〉<br>
 * 〈功能详细描述〉
 *
 * @author 16080951(林健)
 * @Copyright (c) 2016 苏宁软件技术有限公司 版权所有
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
@Service
public class ElasticSearchServiceImpl extends BaseService implements ElasticSearchService {

    /**
     * 查询所有结果(有可能不准，因为按评分规则来排序 ，用的是search，而不是scroll)
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> List<T> queryForList(String jsonId, Class<T> clazz, Map<String, Object> params)
            throws Exception {
        List<T> result = new ArrayList<T>();
        try {
            //查询语句
            String queryJson = EsConfig.getEsQueryConfig(jsonId).getQueryString();
            //查询索引
            String index = EsConfig.getEsQueryConfig(jsonId).getIndex();
            //查询类型
            String type = EsConfig.getEsQueryConfig(jsonId).getType();

            //若变量不变空，则替换里面的值。否则直接执行。
            if (MapUtils.isNotEmpty(params)) {
                Iterator paramsIterator = params.entrySet().iterator();
                while (paramsIterator.hasNext()) {
                    Map.Entry entry = (Map.Entry) paramsIterator.next();
                    String keys = (String) entry.getKey();
                    Object values = null;
                    if (entry.getValue() instanceof Number) {
                        values = (Number) entry.getValue();
                    } else {
                        values = (String) entry.getValue();
                    }
                    queryJson = queryJson.replaceAll("#" + keys + "#", "\"" + values + "\"");
                }
            }

            QueryBuilder queryStringJSONBuilder = new QueryStringJSONBuilder(JSONObject.fromObject(queryJson));
            Search search = new Search.Builder(queryStringJSONBuilder.toString())
                    .addIndex(index)
                    .addType(type)
                    .setParameter(Parameters.SIZE, MAX_LIST_CAPACITY)
                    .setParameter("timeout", TimeValue.timeValueMillis(SCROLL_TIMEOUT))
                    .setSearchType(QUERY_THEN_FETCH).build();
            SearchResult searchResult = JestUtils.getInstance().execute(search);
            //总量
            long total = ((SearchResult) searchResult).getTotal();
            //封装对象
            packageObjectPojo(total, clazz, result, searchResult);
            return result;
        } catch (IOException e) {
            logger.error("es queryForList error:", ExceptionUtils.getFullStackTrace(e));
            return result;
        }
    }

    /**
     * 按分页查询
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> List<T> queryForPage(String jsonId, Class<T> clazz, Map<String, Object> params)
            throws Exception {
        List<T> result = new ArrayList<T>();
        //取出客户端
        JestClient client = JestUtils.getInstance();
        try {
            //查询语句
            String queryJson = EsConfig.getEsQueryConfig(jsonId).getQueryString();
            //查询索引
            String index = EsConfig.getEsQueryConfig(jsonId).getIndex();
            //查询类型
            String type = EsConfig.getEsQueryConfig(jsonId).getType();
            //替换查询串中的##
            queryJson = replaceQueryJsonVar(params, queryJson);

            //取出每页多少条
            int pageSize = MapUtils.getIntValue(params, Constants.CONSTANTS_PAGE_SIZE, Constants.PAGE_SIZE);

            pageSize = pageSize <= Constants.SHARDS ? Constants.SHARDS : pageSize;

            //当前页
            int currentPage = MapUtils.getIntValue(params, Constants.CONSTANTS_CURRENT_PAGE, Constants.CURRENT_PAGE);

            //序列化json
            //QueryBuilder queryStringJSONBuilder = new QueryStringJSONBuilder(JSONObject.fromObject(queryJson));

            //执行搜索
            Search scrollSearch = new Search.Builder(queryJson).addIndex(index)
                    .addType(type)
                    .setParameter(Parameters.SIZE, pageSize)
                    .setParameter(Parameters.SCROLL, "5m")
                    .setSearchType(DFS_QUERY_THEN_FETCH).build();
            JestResult scrollResult = client.execute(scrollSearch);

            if (!scrollResult.isSucceeded())
                return result;
            do {
                //总量
                long total = ((SearchResult) scrollResult).getTotal();

                //共多少页
                long totalPage = (total % pageSize == 0) ? total / pageSize : (total / pageSize) + 1;

                //若实际的页数不够总页数，默认以总页放入。
                currentPage = (int) (currentPage >= totalPage ? totalPage : currentPage);

                String scrollId = scrollResult.getJsonObject().get("_scroll_id").getAsString();

                for (int i = 1; i <= totalPage; i++) {
                    if (i == currentPage) {
                        packageObjectPojo(total, clazz, result, scrollResult);
                        return result;
                    }

                    //否则继续查询
                    SearchScroll scroll = new SearchScroll.Builder(scrollId, "5m").build();
                    scrollResult = client.execute(scroll);
                }

                return result;

            } while (StringUtils.isNotEmpty(scrollResult.getSourceAsString()));

        } catch (Exception e) {
            logger.error("es queryForPage error:", ExceptionUtils.getFullStackTrace(e));
            return result;
        }
    }

    /**
     * 替换##等常用字符为指定常量。
     * @param params
     * @param queryJson
     * @return
     */
    private String replaceQueryJsonVar(Map<String, Object> params, String queryJson) {
        if (MapUtils.isNotEmpty(params)) {
            Iterator paramsIterator = params.entrySet().iterator();
            while (paramsIterator.hasNext()) {
                Map.Entry entry = (Map.Entry) paramsIterator.next();
                Object values = null;
                if (entry.getValue() instanceof Number) {
                    values = (Number) entry.getValue();
                } else {
                    values = (String) entry.getValue();
                }
                String keys = (String) entry.getKey();

                queryJson = queryJson.replaceAll("#" + keys + "#", "\"" + values + "\"");
            }
        }
        return queryJson;
    }

    /**
     * 根据条件删除
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return true:success,false:fail
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> boolean deleteByQuery(String jsonId, Class<T> clazz, Map params) throws Exception {
        Client client = EsUtils.openConnection();
        try {
            //查出匹配的结果
            List<T> result = queryForList(jsonId, clazz, params);
            for (T t : result) {
                JestUtils.deleteDoc(t.get_id(), t.get_index(), t.get_type());
            }
            return true;
        } catch (Exception e) {
            logger.error("es deleteByQuery error:", ExceptionUtils.getFullStackTrace(e));
            return false;
        }
    }

    /**
     * 更新数据（_id,_type,_index)必传。
     *
     * @param date      需要插入或更新的数据
     * @param indexName 索引名
     * @param type      类型
     * @return true:成功，false失败
     * @throws Exception the exception
     */
    @Override
    public <T extends EsBasePojo> boolean insertOrUpdate(List<T> date, String indexName, String type) throws Exception {
        try {
            if (CollectionUtils.isEmpty(date))
                return false;
            for (T t : date) {
                JestUtils.insertOrUpdateDoc(t.get_id(), t, indexName, type);
            }
            return true;
        } catch (Exception e) {
            logger.error("es insertOrUpdate error:", ExceptionUtils.getFullStackTrace(e));
            return false;
        }
    }
}
